package tailfile

import (
	"Logger/logAgent/kafka"
	"context"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/hpcloud/tail"
	"strings"
	"time"
)

//var tailObj *tail.Tail

type TailTask struct {
	tailObj  *tail.Tail
	Topic    string
	Filename string

	ctx    context.Context
	cancel context.CancelFunc
}

func newTailTask(topic, filename string) (*TailTask, error) {
	config := tail.Config{
		ReOpen:    true,
		Follow:    true,
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
		MustExist: false,
		Poll:      true,
	}
	tailObj, err := tail.TailFile(filename, config)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithCancel(context.Background())
	task := &TailTask{
		tailObj:  tailObj,
		Topic:    topic,
		Filename: filename,
		ctx:      ctx,
		cancel:   cancel,
	}

	return task, err
}

func (t *TailTask) run() {
	//读取数据
	var (
		msg *tail.Line
		ok  bool
	)
	for {
		select {
		case msg, ok = <-t.tailObj.Lines: //chan tail.line
			if !ok {
				fmt.Println("Tail file close reopen,filename:", t.tailObj.Filename)
				time.Sleep(1 * time.Second)
				continue
			}
			if len(strings.Trim(msg.Text, "\r")) == 0 {
				continue
			}
			//fmt.Println("msg:", msg.Text)
			kafkaMsg := &sarama.ProducerMessage{
				Topic: t.Topic,
				Value: sarama.StringEncoder(msg.Text),
			}
			kafka.SendMessage(kafkaMsg)
		case <-t.ctx.Done(): //若收到取消信号，则关闭协程
			return

		}

	}
}

//func Run(topic string) {
//	//读取数据
//	var (
//		msg *tail.Line
//		ok  bool
//	)
//	for {
//		msg, ok = <-tailObj.Lines //chan tail.line
//		if !ok {
//			fmt.Println("Tail file close reopen,filename:", tailObj.Filename)
//			time.Sleep(1 * time.Second)
//			continue
//		}
//		if len(strings.Trim(msg.Text, "\r")) == 0 {
//			continue
//		}
//		//fmt.Println("msg:", msg.Text)
//		kafkaMsg := &sarama.ProducerMessage{
//			Topic: topic,
//			Value: sarama.StringEncoder(msg.Text),
//		}
//		kafka.SendMessage(kafkaMsg)
//	}
//
//}
